package com.hadoop.sort;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class SortMapper extends
		Mapper<LongWritable, Text, DateKey, IntWritable> {
	private final static IntWritable outputvalue = new IntWritable(1);
	private DateKey outputkey = new DateKey();
	
	public void map(LongWritable key, Text value, Context context){
		
		if(key.get() > 0){
			
			String[] columns = value.toString().split(",");
			if(columns != null && columns.length > 0){
				try{

					
					outputkey.setYear(columns[0]);
					outputkey.setMonth(new Integer(columns[1]));		
					
					int firstData = Integer.parseInt(columns[2]);
					int secondData = Integer.parseInt(columns[3]);
					
					outputvalue.set(firstData + secondData);
					
					context.write(outputkey, outputvalue);
					
				}catch(Exception e){
				}
			}
		}
	}
}
